package com.freez.spark.core.operation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 *
 * @Classname: ComplexOperation.java
 * @Author: zcs
 * @Date: 2021年-12月-08日 周三 11:18
 * @Description: 复杂聚合类算子
 */
public class ComplexOperation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("SparkCore-ComplexOperation").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaPairRDD<String, String> aKVRDD = jsc.parallelizePairs(Arrays.asList(
                new Tuple2<String, String>("a", "20"),
                new Tuple2<String, String>("math", "40"),
                new Tuple2<String, String>("a", "30"),
                new Tuple2<String, String>("Chinese", "60"),
                new Tuple2<String, String>("math", "50"),
                new Tuple2<String, String>("Chinese", "70")
        ), 3);
        /**
         * aggregateByKey对JavaPairRDD中相同key的元素进行聚合运算，返回相应的JavaPairRDD,其参数有三：
         * 1). zeroValue：进行分区内计算的默认值；
         * 2). seqFunc：分区内计算逻辑；
         * 3). combFunc：分区间的计算
         */
        JavaPairRDD<String, String> aggregateRDD = aKVRDD.aggregateByKey("4", new Function2<String, String, String>() {
            @Override
            public String call(String v1, String v2) throws Exception {
                return v1 + v2;
            }
        }, new Function2<String, String, String>() {
            @Override
            public String call(String v1, String v2) throws Exception {
                return v1 + "_" + v2;
            }
        });
        aggregateRDD.foreach(new VoidFunction<Tuple2<String, String>>() {
            @Override
            public void call(Tuple2<String, String> s) throws Exception {
                System.out.println(s);
            }
        });
        /**
         * fold foldByKey 类似于aggregate()和aggregateByKey()，即fold()其实是将aggregate()中的seqOp操做和combOp操做结合在一起了
         */
        JavaPairRDD<String, String> foldByKeyRDD = aKVRDD.foldByKey("8", new Function2<String, String, String>() {
            @Override
            public String call(String v1, String v2) throws Exception {
                return v1 + v2;
            }
        });
        foldByKeyRDD.foreach(new VoidFunction<Tuple2<String, String>>() {
            @Override
            public void call(Tuple2<String, String> s) throws Exception {
                System.out.println(s);
            }
        });
        jsc.close();
    }
}